Solves #68683

Airflow version observed 3.2.1

KubernetesExecutor.sync() re-runs _change_state() over the entire self.completed set, and nothing ever removes entries from that set.
Iteration over the self.completed is nested inside the result-queue while True

With delete_worker_pods=False, the data structure only grows, with no removals. So every adopted completed pod is re-PATCHed forever, and the set grows monotonically over the scheduler's lifetime.
With delete_worker_pods=True, the same happens with pod deletion.

The same pod name is deleted many times within seconds:

2026-06-11T18:49:36.108968Z  Deleting pod trigger-test-dags-trigger-zwyr4icn ...
2026-06-11T18:49:36.228925Z  Deleting pod trigger-test-dags-trigger-zwyr4icn ...
2026-06-11T18:49:36.397320Z  Deleting pod trigger-test-dags-trigger-zwyr4icn ...
... (166 total for this one pod)

This starves the scheduler loop: with (in my case ~1,855) scheduled TIs waiting, most of each loop is spent re-deleting finished pods instead of launching new ones.

Expected behaviour

Each finished worker pod should be patched or deleted once. Subsequent scheduler loops should not re-issue calls for pods already processed.

Reproduce

1. Deploy Airflow 3.2.x with KubernetesExecutor.

2. Trigger DAGs with many mapped tasks (e.g. 100+ mapped trigger tasks) so pods complete in quick succession.

3. Inspect scheduler logs:

   kubectl logs <scheduler-pod> -c scheduler | grep "Deleting pod" \
     | sed 's/.*Deleting pod \([^ ]*\) in.*/\1/' | sort | uniq -c | sort -rn | head

4. Observe the same pod names with delete counts >> 1.

RCA

Introduced/regressed around #55797 (Oct 2025), which added a self.completed set for orphaned completed pod adoption.
In KubernetesExecutor.sync() (providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/executors/kubernetes_executor.py):

1. self.completed is processed inside the while True loop that drains result_queue — for every completion event, all entries in self.completed call _change_state() again, each triggering delete_pod():

   while True:
       results = self.result_queue.get_nowait()
       ...
       self._change_state(results)
   
       for result in self.completed:   # <-- nested inside while True
           self._change_state(result)

2. self.completed is never cleared after processing, so entries accumulate and are re-processed on every subsequent completion event.

Expected delete volume ≈ num_result_queue_events × (1 + len(self.completed))

Proposed fix

1. Move for result in self.completed outside the result-queue drain loop (once per sync()).
2. Clear or discard from self.completed after successful processing.
3. Deduplicate by pod_name when adopting completed pods.

Expected impact

Before: delete_calls ≈ num_result_events × (1 + len(completed)) per sync().
After: delete_calls ≈ num_result_events + len(completed) per sync(), with completed cleared after processing.

Was generative AI tooling used to co-author this PR?

• Yes
  Generated-by: [Cursor Composer ] following the guidelines